package ex.other;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import java.util.concurrent.ExecutionException;


public class CommApiTestKafka {
    public static Table report(Table transactions) {
        throw new RuntimeException();
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //        env.setParallelism(1);
        // 1     	 

        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .build();

        TableEnvironment tableEnv = TableEnvironment.create(settings);

        // 2 

        String createInDDL =
                "CREATE TABLE inputTable (" +
                "  `userid` BIGINT," +
                "  `itemid` BIGINT," +
                "  `behavior` STRING" +
                ") WITH (" +
                "  'connector' = 'kafka'," +
                "  'topic' = 'fuser_behavior'," +
                "  'properties.bootstrap.servers' = '192.168.1.6:9092'," +
                "  'properties.group.id' = 'testGroup'," +
                "  'scan.startup.mode' = 'earliest-offset'," +
                "  'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createInDDL);

        // 3  
        String createOutDDL = "CREATE TABLE outPutTable (" +
                "userid BIGINT, " +
                "itemid BIGINT, " +
                "behavior STRING " +
                ") WITH (" +
                "   'connector'  = 'jdbc'," +
                "   'url'        = 'jdbc:mysql://localhost:3306/test?characterEncoding=utf8&serverTimezone=UTC&useSSL=false'," +
                "   'table-name' = 'userbehavior'," +
                "   'driver'     = 'com.mysql.cj.jdbc.Driver'," +
                "   'username'   = 'root'," +
                "   'password'   = 'liji@2024'," +
                "   'sink.buffer-flush.max-rows' = '0' " +
                ")";
        tableEnv.executeSql(createOutDDL);

        // 4 
        Table inputTable = tableEnv.from("inputTable");
        inputTable.printSchema();

        // 5 select user_name,ts as totalTs from clickTable where ts>500
        Table inputTableResult = inputTable.limit(100);
        inputTableResult.printSchema();
        // 6

        tableEnv.createTemporaryView("inputTableView", inputTableResult);
        Table resultTable2 = tableEnv.sqlQuery("select userid,itemid,behavior from inputTableView");
        resultTable2.printSchema();

        /**7.1-a 7.1-b   **/

        // 7.1-a 
//        resultTable2.executeInsert("userbehavior");
        // 7.1-b  
        tableEnv.executeSql("insert into userbehavior select *  from inputTableView"); //

        String createPrintOutDDL = "CREATE TABLE printOutTable (" +
                "userid BIGINT, " +
                "itemid BIGINT, " +
                "behavior STRING " +
                ") WITH (" +
                " 'connector' = 'print' " +
                ")";
        tableEnv.executeSql(createPrintOutDDL);
        //
        resultTable2.executeInsert("printOutTable");


    }
}

